package com.arica.mq.consumer.listener;

import com.alibaba.fastjson.JSONException;
import com.arica.framework.common.result.Result;
import com.arica.framework.dto.MqConsumptionDTO;
import com.arica.mq.consumer.processor.MessageProcessor;
import com.arica.mq.consumer.processor.ProcessHelper;
import com.arica.mq.consumer.processor.ProcessorFactory;
import com.arica.project.mq.service.MqConsumptionService;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.text.MessageFormat;
import java.util.List;
import java.util.Objects;

@Component
public class BaseMessageListener implements MessageListenerConcurrently {
    private static final Logger logger = LoggerFactory.getLogger(BaseMessageListener.class);

    // 该 BaseMessageListener 所属的 BaseConsumerBean，用于在该 BaseMessageListener 中引用对应的 BaseConsumerBean 中的属性
    private BaseConsumerBean baseConsumerBean;

    // 该 BaseMessageListener 所属的 BaseConsumerBean 对应的消费组名称
    private String groupId;

    @Autowired
    private MqConsumptionService mqConsumptionService;

    @Autowired
    private ProcessHelper processHelper;

    public BaseMessageListener() {
    }

    private String getSentinelResourceName(Message message) {
        // 如果没有tag，则以 groupId:topic 为资源名作为埋点
        String resourceName = groupId + ":" + message.getTopic();
        if (StringUtils.isNotBlank(message.getTags())) {
            resourceName = resourceName + ":" + message.getTags();
        }

        return resourceName;
    }

    public BaseConsumerBean getBaseConsumerBean() {
        return baseConsumerBean;
    }

    public void setBaseConsumerBean(BaseConsumerBean baseConsumerBean) {
        this.baseConsumerBean = baseConsumerBean;
    }

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        // TODO: 2019-11-08 改为 ApplicationListener
        if (StringUtils.isBlank(groupId)) {
            groupId = baseConsumerBean.getCommonProperties().getGroup();
        }

        MessageExt message = list.get(0);

        String messageBody = new String(message.getBody());
//        Entry entry = null;
//        try {
//            String resourceName = getSentinelResourceName(message);
//            entry = SphU.entry(resourceName);

            List<MessageProcessor> processorList = ProcessorFactory.getProcessors(message.getTopic(), message.getTags());
            if (CollectionUtils.isEmpty(processorList)) {
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }

            for (MessageProcessor processor : processorList) {
                MqConsumptionDTO mqConsumptionDTO = null;
                try {
                    String sourceClassName = processHelper.getSourceClassName(processor, message.getTopic(), message.getTags());
                    if (StringUtils.isBlank(sourceClassName)) {
                        continue;
                    }
                    Object sourceData = processHelper.getBusinessData(messageBody, sourceClassName);
                    Object targetData = processor.convertData(sourceData);
                    boolean isIdempotent = true;
                    if (processor.needIdempotent()) {
                        Result<MqConsumptionDTO> result = processHelper.isIdempotent(targetData, message, processor);
                        isIdempotent = result.isStatus();
                        mqConsumptionDTO = result.getEntry();
                    }
                    if (isIdempotent) {
                        processor.process(targetData);
                    }
                } catch (JSONException jsonException) {
                    logger.error("json data error", jsonException);
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                } catch (Exception e) {
                    logger.error(MessageFormat.format("catch exception BaseMessageListener message={0},body={1},tag={2},processKey={3})",
                            message, new String(message.getBody()), message.getTags(), processor.getProcessType()), e);

                    if (Objects.isNull(mqConsumptionDTO)) {
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                    //如果幂等了，但是业务处理失败，需要删除记录
                    if (mqConsumptionDTO != null && StringUtils.isNotBlank(mqConsumptionDTO.getMqKey())) {
                        mqConsumptionService.deleteByKey(mqConsumptionDTO.getMqKey());
                    }

                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//        } catch (BlockException ex) {
//            // 被限流时，重新消费(后续考虑是否根据策略做不同处理，比如：直接丢弃，或重新消费)
//            return Action.ReconsumeLater;
//        } catch (Exception e) {
//            logger.error("consume error, message={}, body={}", message, new String(message.getBody()), e);
//            return Action.ReconsumeLater;
//        } finally {
//            if (entry != null) {
//                entry.exit();
//            }
//        }
    }
}
